//
// Created by Yiming Zhang on 11/6/18.
//

//#include <stdlib.h>
#include <cstdlib>
#include <string.h>
#include <time.h>
#include <errno.h>
#include <limits.h>
#include <dirent.h>
#include <sys/socket.h>
#include <sys/statvfs.h>
#include <arpa/inet.h>

#include "st.h"
#include "redis-client.h"
#include "protocol.h"

#define USE_LOCAL_LOG_LEVEL
#include "log.h"
#include "heartbeat.h"
#include "ursax.h"
#include "utcache.h"
#include "networking.h"
#include "backend-fs.h"
#include "statistics.h"
#include "master-request.h"
#include "concurrent-connection-pool.h"
//#include "pooled-functions.h"

#define UPDATE_INTERVAL (10*60)
#define UPDATE_TIMES (50)

int g_disk_type = URSAX_DISK_TYPE_HDD;
uint64_t bigjournal_size = 0;
uint64_t ecjournal_size = 0;
bool enable_ec_auto_flush = false;
uint64_t max_ec_journal_size = 100 * ONE_GB;

struct Operation
{
    uint64_t chunkid;
    struct Operation* next;
    int op;
};

struct ChunkServer
{
    int n_control_code;
    char control_code[1024];
    struct Operation* op_journal;
    struct foo_hashset* all_chunks;
    struct ChunkServerInfo* csi;
    st_thread_t heartbeat_thread;
    bool sleeping, running, updating_journal, updating_full;
};

static struct ChunkServer _cs;
static struct ChunkServer* cs = &_cs;
struct ChunkServerInfo *g_csi;

static inline int list_add_chunk(uint64_t chunkid)
{
    return foo_hashset_put(cs->all_chunks, chunkid);
}

static inline int list_remove_chunk(uint64_t chunkid)
{
    return foo_hashset_remove(cs->all_chunks, chunkid);
}

static int get_data_dir_space(const char *data_dir, uint64_t *capacity, uint64_t *free)
{
    struct statvfs buf;
    int ret = statvfs(data_dir, &buf);
    if (ret < 0) {
        LOG_ERROR("statvfs(%s) failed: (%d, %s)", data_dir, errno, strerror(errno));
        return ret;
    }

    *capacity = buf.f_blocks * buf.f_bsize;
    // reserved 5% like `df` shows, see more at:
    // http://unix.stackexchange.com/questions/25602/why-is-free-space-used-space-total-size-in-df
    int64_t current_free = (buf.f_bfree - buf.f_blocks/20) * buf.f_bsize;
    if (0 > current_free) {
        LOG_WARN("[%s:%d] free space runs out, set to zero!!!, last: %lld, current: %lld",
                 __func__, __LINE__, *free, current_free);
        *free = 0;
    }
    else {
        *free = current_free;
    }

    return 0;
}

static int get_chunkserver_space(const char *data_dir, uint64_t *capacity, uint64_t *free)
{
    if (unlikely(data_dir == NULL || capacity == NULL || free == NULL)) {
        LOG_ERROR("Input params error");
        return -1;
    }
    int ret = get_data_dir_space(data_dir, capacity, free);
    if (unlikely(ret < 0)) {
        LOG_ERROR("get_data_dir_space error");
        return -1;
    }
//    ret = get_raw_space(capacity, free);
//    if (unlikely(ret < 0)) {
//        LOG_ERROR("get_raw_space error");
//        return -1;
//    }
    return 0;
}

void set_big_journal_size(uint64_t size)
{
    bigjournal_size = size;
}

static int update_server_info()
{
    char loadavg[128];

    read_file("/proc/loadavg", loadavg, sizeof(loadavg));
    for (uint32_t i=0; i<sizeof(loadavg); ++i)
    {
        if (unlikely(loadavg[i] == ' '))
            loadavg[i] = ',';
        if (unlikely(loadavg[i] == '\n'))
            loadavg[i] = '\0';
        if (unlikely(loadavg[i] == '\0'))
            break;
    }

    char cmd[2048], *p = cmd;
    char *end = cmd + sizeof(cmd);
    struct ChunkServerInfo* csi = cs->csi;

    // when chunkserver is running, we chdir to chunkserver's data dir
    get_chunkserver_space("./", &csi->capacity, &csi->free);
    p += snprintf(p, end - p, " HMSET %s",				chunk_server_id);
    p += snprintf(p, end - p, " ip %s",					csi->ip);
    p += snprintf(p, end - p, " port %u",				csi->port);
    p += snprintf(p, end - p, " rootdir %s",			csi->root);
    p += snprintf(p, end - p, " capacity %llu",			(long long int)csi->capacity);
    p += snprintf(p, end - p, " free %llu",				(long long int)csi->free);
    p += snprintf(p, end - p, " total_chunk_size %llu", (long long int)csi->total_chunk_size);
    p += snprintf(p, end - p, " loadavg %s",			loadavg);
    p += snprintf(p, end - p, " hostname %s",			csi->hostname);
    p += snprintf(p, end - p, " disktype %d", 			g_disk_type);
    p += snprintf(p, end - p, " bigjournalsize %ld", 	bigjournal_size);
    p += snprintf(p, end - p, " ecjournalsize %ld", 	ecjournal_size);

    ursax_statis_report_all(p, end - p);
    int ret = redis_execute(cmd);
    return ret;
}

static int _update_chunk(uint64_t key, void* data, void* arg)
{
    LOG_DEBUG("enter %s key %ld data %ld ", __func__, key , data);
    (void)data; (void)arg;
    master_add_chunk(key);
    return 0;
}

static void* _update_chunks_full(void* p)
{
    LOG_DEBUG("enter %s", __func__);
    bool clear_before_do = (bool)p;
    cs->updating_full = true;

    if(clear_before_do){
        LOG_INFO("_update_chunks_full clear before update !");
        master_clear_chunks();
    }
    LOG_INFO("update all chunks to redis start");
    foo_hashset_iterate(cs->all_chunks, &_update_chunk, NULL);
    LOG_INFO("update all chunks to redis end");

    cs->updating_full = false;
    return NULL;
}

static int update_chunks_full(bool create_thread, bool clear_before_do)
{
    if (cs->updating_full){
        LOG_ERROR("updating full is true ret");
        return 0;
    }

    if (create_thread){
        st_thread_create(_update_chunks_full, (void*)clear_before_do, 0, ST_STACK_SIZE);
    } else {
        _update_chunks_full(NULL);
    }
    return 0;
}

static int master_check_and_refill()
{
    LOG_INFO("enter %s", __func__);
    struct redisReply* r = redis_execute_result("SISMEMBER ChunkServers %s", chunk_server_id);
    if (r == NULL){
        LOG_ERROR("%s ERROR AGANIN", __func__);
        return -1;
    }
    // todo check add ===> (r->type > REDIS_REPLY_ERROR || r->type <REDIS_REPLY_STRING ) ||
    bool ignorant = (r->type != REDIS_REPLY_INTEGER) || // todo error here
                    (likely(r->type == REDIS_REPLY_INTEGER) && unlikely(r->integer == 0)); // fixme sometimes,type == 30359776
    // bool ignorant = true;
    LOG_DEBUG("%s ignorant %d r->type == REDIS_REPLY_INTEGER %d type %d r->integer %d",
              __func__, ignorant, r->type == REDIS_REPLY_INTEGER, r->type, r->integer);
    freeReplyObject(r);

    if (ignorant)
    {
        master_add_server();
        update_server_info();
        update_chunks_full(true, true);
        keep_alive(cs->csi->interval + MIN(cs->csi->interval, 10));
    }
    LOG_INFO("leave %s", __func__);
    return 0;
}

static void* _update_chunks_journal(void *__p)
{
    (void)__p;
    LOG_DEBUG("%s ", __func__);
    cs->updating_journal = true;

    struct Operation* p;
    go_on_update:
    p = cs->op_journal;
    cs->op_journal = NULL;
    while(p)
    {
        if (p->op == CHUNK_CREATE)
        {
            LOG_DEBUG("%s add redis %lx", __func__, p->chunkid);
            master_add_chunk(p->chunkid);
            list_add_chunk(p->chunkid);
        }
        else if (p->op == CHUNK_DELETE)
        {
            LOG_DEBUG("%s del redis %lx", __func__, p->chunkid);
            master_remove_chunk(p->chunkid);
            list_remove_chunk(p->chunkid);
        }
        else
        {
            LOG_ERROR("invalid journal code %d", p->op);
        }
        void* tmp = p;
        p = p->next;
        free(tmp);
    }
    if(cs->op_journal){// when updating some, some other have been add
        LOG_INFO("when updating some, some other have been add, go on update");
        goto go_on_update;
    }

    cs->updating_journal = false;
    return 0;
}

static void update_chunks_journal()
{
    if(unlikely(cs->updating_journal))
        return;

    cs->updating_journal = true;
    st_thread_create(_update_chunks_journal, NULL, 0, ST_STACK_SIZE);
}

static int process_control_codes()
{
    while(cs->n_control_code > 0)
    {
        cs->n_control_code--;
        char code = cs->control_code[ cs->n_control_code ];
        switch(code) {
            case HB_CONTROL_CODE_UPDATE:
                update_chunks_journal();
                break;

            case HB_CONTROL_CODE_FULL_UPDATE:
                update_chunks_full(true, false);
                break;

            case HB_CONTROL_CODE_EXIT:
                cs->n_control_code = 0;
                return -1;

            case HB_CONTROL_CODE_FILL_REDIS:
                master_check_and_refill();
                break;

            case HB_CONTROL_CODE_STOP:
                LOG_INFO("[%s:%d] chunkserver restart", __func__, __LINE__);
                return -2;

            default:
                LOG_ERROR("invalid control %u", code);
                break;
        }
    }
    return 0;
}

//static inline void check_and_flush_self_all_ec_chunks()
//{
//    if(!enable_ec_auto_flush) {
//        return;
//    }
//    if(ecjournal_size < max_ec_journal_size) {
//        return;
//    }
//    LOG_INFO("max_ec_journal_size %ld ecjournal_size is %ld start flush self", max_ec_journal_size, ecjournal_size);
//    pooled_chunk_flush_ec_journal(cs->csi->ip_int, cs->csi->port, -1, -1);
//}

void heartbeat_fullupdate()
{
    master_add_server();
    update_server_info();
    heartbeat_add_control_code(HB_CONTROL_CODE_FULL_UPDATE);
}

bool update_full = false;

static void* heartbeat(void* arg)
{
    (void)arg;
    cs->running = true;
    master_add_server();
    update_server_info();
    update_chunks_full(false, true);
    if (cs->csi->request_server_up) {
        request_manager_chunkserver_up(cs->csi->ip_int, cs->csi->port);
    }
#ifdef USE_TIME
    time_t last_update = time(0);
#else
    int little_update_tick = 0;
#endif

    int ret = 0;
    while(true) {
        LOG_INFO("heartbeating...");
        master_add_server();
        keep_alive(cs->csi->interval + MIN(cs->csi->interval, 10));
        cs->sleeping = true;
        time_t before_sleep = time(0);
        int ret_sleep = st_sleep(cs->csi->interval);
        cs->sleeping = false;
#ifdef USE_TIME
        check_time_update(&last_update);
#else
        if(time(0) - before_sleep > 5 || ret_sleep == 0){
            ++little_update_tick;
        }
        if(little_update_tick >= UPDATE_TIMES || update_full ){
            LOG_INFO("start update all chunks");
            update_full = false;
            heartbeat_fullupdate();
            check_manager_in_redis();
            little_update_tick = 0;
        }
#endif
        ret = process_control_codes();
        if(unlikely(ret < 0)){
            break;
        }
        update_chunks_journal();
        update_server_info(); // update capacity info to master periodly
//        check_and_flush_self_all_ec_chunks();
    }
    while(cs->updating_full || cs->updating_journal){
        LOG_INFO("cs->updating_full || cs->updating_journal WAIT!!!");
        st_usleep(100*1000);
    }

    /**< keep meta data in restart case which returns -2 */
    if (-1 == ret) {
        master_clear_chunks();
        master_remove_server();
    }
    cs->running = false;
    return NULL;
}

static void heartbeat_on_redis_connect()
{
    static int n_connection = 0;
    if (n_connection++ > 0)	{ // not the first connection
        // heartbeat_add_control_code(HB_CONTROL_CODE_FILL_REDIS);
        // heartbeat_wake_up();
        LOG_DEBUG("heartbeat_on_redis_connect %d", n_connection);
        master_check_and_refill();
    }
    LOG_DEBUG("heartbeat_on_redis_connect %d", n_connection);
}

int heartbeat_init(struct ChunkServerInfo* csi)
{
//    register_local_log_level("s.heartbeat");

    LOG_INFO("heartbeat_init");

    int ret = -1;
    cs->csi = csi;
    g_csi = csi;
    cs->all_chunks = foo_hashset_create();

    int size;
    char ip[32];
    struct ChunkID* chunks = NULL;
    ret = be_chunk_list(-1, -1, &chunks);
    if (unlikely(ret < 0)) {
        LOG_ERROR("list chunks error");
        free(chunks);
        goto error1;
    }
//    be_chunk_list_recycles_raw_fs();

    size = ret;
    for (int i=0; i<size; ++i) {
        list_add_chunk(chunks[i].bits);
    }
    free(chunks);

    strcpy(ip, csi->ip);
    if(strcmp(ip, "0.0.0.0") == 0 || ip[0] == '\0') {
        ret = gethostbypeer_str(redis_ip, ip);
        if(ret < 0) {
            LOG_FATAL("unable to determin ip and port, exit");
            exit(-1);
        }
    }
    parse_ip(ip, &csi->ip_int);
    snprintf(chunk_server_id, sizeof(chunk_server_id), "%s:%u", ip, csi->port);
    LOG_INFO("ChunkServerID: %s", chunk_server_id);

    on_redis_connect = &heartbeat_on_redis_connect;

    cs->heartbeat_thread = st_thread_create(&heartbeat, NULL, 0, ST_STACK_SIZE);
    if (unlikely(cs->heartbeat_thread == NULL)) {
        LOG_ERROR("heartbeat thread creation failed");
        ret = -1;
        goto error1;
    }

    return 0;

error1:
    foo_hashset_delete(cs->all_chunks);
    cs->all_chunks = NULL;
    return ret;
}

int heartbeat_add_control_code(int code)
{
    while (cs->n_control_code >= (int)sizeof(cs->control_code))
    {
        LOG_DEBUG("cs->n_control_code >= (int)sizeof(cs->control_code)");
        heartbeat_wake_up();
        st_usleep(1000);
    }
    cs->control_code[ cs->n_control_code++ ] = code;
    return 0;
}

int heartbeat_wake_up()
{
    if (cs->sleeping)
        st_thread_interrupt(cs->heartbeat_thread);
    return 0;
}

int heartbeat_fini(bool keep_meta)
{
    int code = keep_meta ? HB_CONTROL_CODE_STOP : HB_CONTROL_CODE_EXIT;
    if (unlikely(!cs->running))
        return 0;
    heartbeat_add_control_code(code);
    heartbeat_wake_up();
    while(cs->running){
        LOG_DEBUG("cs->running, WAIT heartbeat over");
        st_usleep(1000*1000);
    }

    foo_hashset_delete(cs->all_chunks);
    cs->all_chunks = NULL;
    return 0;
}

void heartbeat_add_chunk(uint64_t chunkid)
{
    master_add_chunk(chunkid);
    list_add_chunk(chunkid);
}

void heartbeat_remove_chunk(uint64_t chunkid)
{
    // add_journal_op(CHUNK_DELETE, chunkid);
    master_remove_chunk(chunkid);
    list_remove_chunk(chunkid);
}

void get_self_ip_port(uint32_t *ip, uint16_t *port)
{
    *ip = cs->csi->ip_int;
    *port = cs->csi->port;
}